package com.sharp.mq.service;

import com.sharp.kernel.utils.JsonUtil;
import com.sharp.mq.bo.RabbitMqMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.UUID;

/**
 * Title: RabbitMqService
 * Description: Copyright: Copyright (c) 2019 Company: BHFAE
 *
 * @author Sharp
 * @date 2021/7/1 13:05
 */
@Slf4j
@Service
public class RabbitMqService {

    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private AsyncRabbitTemplate asyncRabbitTemplate;

    public void sendMessageByQueue(RabbitMqMessage rabbitMqMessage) {
        CorrelationData correlationData = buildCorrelationData(rabbitMqMessage);
        rabbitTemplate.convertAndSend(rabbitMqMessage.getQueueName(), rabbitMqMessage.getBody(), message -> {
            if (rabbitMqMessage.getDelay() > 0) {
                message.getMessageProperties().setExpiration(String.valueOf(rabbitMqMessage.getDelay()));
                message.getMessageProperties().setHeader("x-delay", rabbitMqMessage.getDelay());
            }
            return message;
        }, correlationData);
    }

    private CorrelationData buildCorrelationData(RabbitMqMessage rabbitMqMessage) {
        return new CorrelationData(UUID.randomUUID().toString());
    }

    public void sendMessage(RabbitMqMessage rabbitMqMessage) {
        CorrelationData correlationData = buildCorrelationData(rabbitMqMessage);
        rabbitTemplate.convertAndSend(rabbitMqMessage.getExchangeName(), rabbitMqMessage.getRouterKey(), rabbitMqMessage.getBody(), message -> {
            if (rabbitMqMessage.getDelay() > 0) {
                message.getMessageProperties().setExpiration(String.valueOf(rabbitMqMessage.getDelay()));
                message.getMessageProperties().setHeader("x-delay", rabbitMqMessage.getDelay());
            }
            return message;
        }, correlationData);
    }

    public void sendMessageFanout(RabbitMqMessage rabbitMqMessage) {
        CorrelationData correlationData = buildCorrelationData(rabbitMqMessage);
        rabbitTemplate.convertAndSend(rabbitMqMessage.getExchangeName(), "", rabbitMqMessage.getBody(), message -> {
            if (rabbitMqMessage.getDelay() > 0) {
                message.getMessageProperties().setExpiration(String.valueOf(rabbitMqMessage.getDelay()));
                message.getMessageProperties().setHeader("x-delay", rabbitMqMessage.getDelay());
            }
            return message;
        }, correlationData);
    }

    public <T> T sendMessageSync(RabbitMqMessage rabbitMqMessage, Class<?> collectionClass, Class<?>... elementClasses) {
        Object obj = rabbitTemplate.convertSendAndReceive(rabbitMqMessage.getExchangeName(), rabbitMqMessage.getRouterKey(), rabbitMqMessage.getBody());
        return JsonUtil.readJson(obj, collectionClass, elementClasses);

    }

    @SuppressWarnings("unchecked")
    public void sendMessageAsync(RabbitMqMessage rabbitMqMessage) {
        AsyncRabbitTemplate.RabbitConverterFuture<Object> future = asyncRabbitTemplate.convertSendAndReceive(rabbitMqMessage.getExchangeName(), rabbitMqMessage.getRouterKey(), rabbitMqMessage.getBody(), message -> {
            if (rabbitMqMessage.getDelay() > 0) {
                message.getMessageProperties().setExpiration(String.valueOf(rabbitMqMessage.getDelay()));
                message.getMessageProperties().setHeader("x-delay", rabbitMqMessage.getDelay());
            }
            return message;
        });
        future.addCallback(rabbitMqMessage.getCallback());
    }
}
